package com.snopzyz.hello.databus.manager;

import com.alibaba.fastjson.JSON;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.snopzyz.hello.databus.domain.BinlogRawData;
import com.snopzyz.hello.databus.domain.Column;
import com.snopzyz.hello.databus.domain.properties.BinlogProperties;
import com.snopzyz.hello.databus.utils.BinlogUtils;
import com.snopzyz.hello.databus.utils.SpringBeanUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.InitializingBean;

import java.io.IOException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class BinlogHandlerManager implements InitializingBean {
    private BinlogProperties properties;
    private final List<BinaryLogClient> clients = new ArrayList<>();
    Map<Long, String> tableId2DbTable = new ConcurrentHashMap<>();
    Map<String, Map<String, Column>> dbTable2Cols = new ConcurrentHashMap<>();


    public BinlogHandlerManager(BinlogProperties properties) {
        this.properties = properties;
    }


    /**
     * Spring Bean 初始化完成后的回调
     * 初始化所有配置的 Binlog 客户端连接，一条 config 就产生一条连接
     */
    @Override
    public void afterPropertiesSet() {
        properties.getConfigs().forEach(this::initializeClient);
    }


    /**
     * 初始化单个 Binlog 客户端连接，根据配置的监听库开启监听
     *
     * @param config 连接配置信息
     */
    private void initializeClient (BinlogProperties.ConnectionConfig config) {
        BinaryLogClient client = new BinaryLogClient(
                config.getHost(),
                config.getPort(),
                config.getUsername(),
                config.getPassword()
        );

        client.setServerId(System.currentTimeMillis() + new Random().nextLong());
        client.registerEventListener(event -> handleEvent(event, config));

        new Thread(() -> {
            try {
                client.connect();
            } catch (IOException e) {
                log.error("Failed to connect to MySQL server", e);
            }
        }).start();
        clients.add(client);

        log.info("成功订阅DTS，config={}", JSON.toJSONString(config));
    }


    /**
     * 接到 MySQL binlog 后的回调事件
     * 1. 通过 MAP_EVENT 维护单个 config 圈定库下的表结构、表id映射关系
     * 2. 对 DML 事件生成更易于处理的 Binlog 实例
     * 3. 将每个 Binlog 实例根据 database 发给对应 config.handler 进行处理
     *
     * @param event binlog事件对象
     * @param config 单条数据库连接配置
     */
    private void handleEvent (Event event, BinlogProperties.ConnectionConfig config) {
        EventHeader header = event.getHeader();
        EventType eventType = header.getEventType();

        // 配出 db-table-col 的关系
        if (eventType == EventType.TABLE_MAP) {
            TableMapEventData tableData = event.getData();
            String database = tableData.getDatabase();
            String table = tableData.getTable();
            // 过滤出仅该 config 关注的
            if (!config.canHandle(database)) {
                return;
            }
            // 做 tableId -> db-table  、 db-table -> 列s
            String dbTable = BinlogUtils.getDbTable(database, table);
            tableId2DbTable.put(tableData.getTableId(), dbTable);
            if (!dbTable2Cols.containsKey(dbTable)) {
                dbTable2Cols.put(dbTable, BinlogUtils.getColMap(database, table));
            }
            return;
        }

        // 提取更易于处理的Binlog对象
        String database = "";
        List<BinlogRawData> binlogRawDataList = new ArrayList<>();
        if (EventType.isWrite(eventType) || EventType.isUpdate(eventType) || EventType.isDelete(eventType)) {
            String dbTable = tableId2DbTable.get(BinlogUtils.getTableId(event.getData()));
            if (StringUtils.isBlank(dbTable)) {
                return;
            }
            database = BinlogUtils.getDb(dbTable);
            binlogRawDataList = BinlogRawData.from(event, dbTable, dbTable2Cols.get(dbTable));
        }

        // 调用连接下配置的 handler#method 处理 DML
        for (BinlogProperties.BinlogHandler configHandler : config.getHandlers()) {
            if (!configHandler.getDatabase().equals(database)) {
                continue;
            }
            invokeHandler(configHandler, binlogRawDataList);
        }
    }


    /**
     * 提取配置中的某个Binlog处理器，处理多条Binlog事件
     * @param configHandler 处理器
     * @param binlogRawDataList binlog事件s
     */
    private void invokeHandler (BinlogProperties.BinlogHandler configHandler, List<BinlogRawData> binlogRawDataList) {
        try {
            Class<?> handlerClass = configHandler.getClazz();

            Object handler = SpringBeanUtils.getBean(handlerClass);
            Method method = handlerClass.getMethod(configHandler.getMethod(), BinlogRawData.class);
            for (BinlogRawData rawData : binlogRawDataList) {
                try {
                    method.invoke(handler, rawData);
                } catch (Exception e) {
                    log.error("Binlog处理器 {}#{} 执行异常", configHandler.getClazz(), configHandler.getMethod(), e.getCause());
                }
            }
        } catch (Exception e) {
            log.error("Binlog配置处理器绑定失败：configHandler={}", JSON.toJSONString(configHandler), e);
        }
    }
}
